package com.aqielife.demo.common.pb;

import lombok.extern.slf4j.Slf4j;

import java.util.concurrent.Flow;
import java.util.concurrent.SubmissionPublisher;
import java.util.concurrent.TimeUnit;

/**
 * @author aqie
 * @date 2022/02/05 13:12
 * @desc 发布-处理-订阅模式
 */
@Slf4j
public class SomeProcessor extends SubmissionPublisher<String> implements Flow.Processor<Integer, String> {

  // 声明订阅令牌
  private Flow.Subscription subscription;

  /**
   * 当订阅关系确立时会触发该方法的执行，即publisher的subscriber()方法的执行
   * 会触发该方法的执行
   * @param subscription
   */
  @Override
  public void onSubscribe(Flow.Subscription subscription) {
    log.info("------>执行订阅者的onSubscribe()方法<------");
    this.subscription = subscription;
    // 首次订阅10条
    this.subscription.request(10);
  }

  /**
   * 对消息的消费过程就在这
   * 该方法的第一次触发是由onSubscribe()方法中的this.subscription.request(10);触发
   * @param item
   */
  @Override
  public void onNext(Integer item) {
    log.info("------>订阅者正在消费该item<------" + item);
    try {
      TimeUnit.SECONDS.sleep(1);
    } catch (InterruptedException e) {
      e.printStackTrace();
    }
    // 订阅者没消费一条消息，就会在订阅5条
    this.subscription.request(5);
  }

  /**
   * 当发布者发布过程中，或订阅关系确立过程中，或者订阅者请求消息过程中，或消费者消费过程中出现异常过程中，
   * 则会触发该方法。
   * @param throwable
   */
  @Override
  public void onError(Throwable throwable) {
    log.info("------>执行onError()方法<------");
    throwable.printStackTrace();
    // 取消订阅关系
    this.subscription.cancel();
  }

  /**
   * 当所有消息消费完毕后会执行该方法
   */
  @Override
  public void onComplete() {
    log.info("------>发布者已关闭，订阅者已将消息全部消费完毕<------");
  }
}
